package site.wuao.simplerxjava.operator;

import android.util.Log;

import java.util.concurrent.TimeUnit;

import rx.Notification;
import rx.Observable;
import rx.Subscriber;
import rx.functions.Action0;
import rx.functions.Action1;

/**
 * 操作符 - 辅助
 *
 * @author wuao
 * @date 2017/12/4
 * @github {https://github.com/CoderWuao}
 * @blogs {http://wuao.site/}
 * @note -
 * ---------------------------------------------------------------------------------------------------------------------
 * @modified -
 * @date -
 * @github -
 * @blogs -
 * @note -
 */
public class HelpOperator {
    /** 标记 */
    private static final String TAG = "HelpOperator";

    /**
     * delay - 延时
     *
     * @note 源Observable发射每项数据之前都暂停一段时间
     */
    public void delay() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        Log.i(TAG, "call_01: " + System.currentTimeMillis());
                        subscriber.onNext(1);
                        Log.i(TAG, "call_02: " + System.currentTimeMillis());
                        subscriber.onCompleted();
                    }
                })
                .delay(2, TimeUnit.SECONDS)
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "onCompleted: " + System.currentTimeMillis());
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: " + System.currentTimeMillis());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "onNext: " + System.currentTimeMillis());
                    }
                });

        // call_01: 1512556227085
        // call_02: 1512556227085
        // onNext: 1512556229089
        // onCompleted: 1512556229090
    }

    /**
     * doOnEach
     *
     * @note Observable每发射一项数据就会调用一次，包括onNext、onError、onCompleted
     */
    public void doOnEach() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        subscriber.onNext(1);
                        subscriber.onNext(2);
                        subscriber.onNext(3);
                        subscriber.onCompleted();
                    }
                })
                .doOnEach(new Action1<Notification<? super Integer>>() {
                    @Override
                    public void call(Notification<? super Integer> notification) {
                        Log.i(TAG, "call: " + notification.getValue());
                    }
                })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "onNext: " + integer);
                    }
                });

        // call: 1
        // onNext: 1
        // call: 2
        // onNext: 2
        // call: 3
        // onNext: 3
        // call: null
        // onCompleted
    }

    /**
     * doOnNext
     *
     * @note 每次执行onNext之前会被调用
     */
    public void doOnNext() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        subscriber.onNext(1);
                        subscriber.onNext(2);
                        subscriber.onNext(3);
                        subscriber.onCompleted();
                    }
                })
                .doOnNext(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.i(TAG, "call: " + integer);
                    }
                })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "onNext: " + integer);
                    }
                });

        // call: 1
        // onNext: 1
        // call: 2
        // onNext: 2
        // call: 3
        // onNext: 3
        // onCompleted
    }

    /**
     * doOnSubscribe
     *
     * @note 当被观察者订阅观察者时会被调用
     */
    public void doOnSubscribe() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        subscriber.onNext(1);
                        subscriber.onNext(2);
                        subscriber.onNext(3);
                        subscriber.onCompleted();
                    }
                })
                .doOnSubscribe(new Action0() {
                    @Override
                    public void call() {
                        Log.i(TAG, "call");
                    }
                })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "onNext: " + integer);
                    }
                });

        // call
        // onNext: 1
        // onNext: 2
        // onNext: 3
        // onCompleted
    }

    /**
     * doOnUnsubscribe
     *
     * @note 1.当被观察者取消订阅观察者时会被调用
     * 2.Observable通过onError或者onCompleted结束后，会取消所有订阅
     */
    public void doOnUnsubscribe() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        subscriber.onNext(1);
                        subscriber.onNext(2);
                        subscriber.onNext(3);
                        subscriber.onCompleted();
                    }
                })
                .doOnUnsubscribe(new Action0() {
                    @Override
                    public void call() {
                        Log.i(TAG, "call");
                    }
                })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "onNext: " + integer);
                    }
                });

        // onNext: 1
        // onNext: 2
        // onNext: 3
        // onCompleted
        // call
    }

    /**
     * doOnCompleted
     *
     * @note 执行onCompleted之前被调用
     */
    public void doOnCompleted() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        subscriber.onNext(1);
                        subscriber.onNext(2);
                        subscriber.onNext(3);
                        subscriber.onCompleted();
                    }
                })
                .doOnCompleted(new Action0() {
                    @Override
                    public void call() {
                        Log.i(TAG, "call");
                    }
                })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "onNext: " + integer);
                    }
                });

        // onNext: 1
        // onNext: 2
        // onNext: 3
        // call
        // onCompleted
    }

    /**
     * doOnError
     *
     * @note 执行onError之前会被调用
     */
    public void doOnError() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        subscriber.onNext(1);
                        subscriber.onNext(2);
                        subscriber.onNext(3);
                        subscriber.onError(new RuntimeException("this is custom exception"));
                    }
                })
                .doOnError(new Action1<Throwable>() {
                    @Override
                    public void call(Throwable throwable) {
                        Log.i(TAG, "call: " + throwable.getMessage());
                    }
                })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "onNext: " + integer);
                    }
                });

        // onNext: 1
        // onNext: 2
        // onNext: 3
        // call: this is custom exception
        // onError: this is custom exception
    }

    /**
     * doOnTerminate
     *
     * @note 当Observable终止，无论是正常还是异常终止之前都会被调用
     */
    public void doOnTerminate() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        subscriber.onNext(1);
                        subscriber.onNext(2);
                        subscriber.onNext(3);
                        subscriber.onError(new RuntimeException("this is custom exception"));
                    }
                })
                .doOnTerminate(new Action0() {
                    @Override
                    public void call() {
                        Log.i(TAG, "call");
                    }
                })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "onNext: " + integer);
                    }
                });

        // onNext: 1
        // onNext: 2
        // onNext: 3
        // call
        // onError: this is custom exception
    }

    /**
     * doAfterTerminate
     *
     * @note 当Observable终止，无论是正常还是异常终止之后都会被调用
     */
    public void doAfterTerminate() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        subscriber.onNext(1);
                        subscriber.onNext(2);
                        subscriber.onNext(3);
                        subscriber.onCompleted();
                    }
                })
                .doAfterTerminate(new Action0() {
                    @Override
                    public void call() {
                        Log.i(TAG, "call");
                    }
                })
                .subscribe(new Subscriber<Integer>() {
                    @Override
                    public void onCompleted() {
                        Log.i(TAG, "onCompleted");
                    }

                    @Override
                    public void onError(Throwable e) {
                        Log.i(TAG, "onError: " + e.getMessage());
                    }

                    @Override
                    public void onNext(Integer integer) {
                        Log.i(TAG, "onNext: " + integer);
                    }
                });

        // onNext: 1
        // onNext: 2
        // onNext: 3
        // onCompleted
        // call
    }

    /**
     * timeout
     *
     * @note 超过了200ms还没有发射数据，就会切换到备用的Observable
     */
    public void timeout() {
        Observable
                .create(new Observable.OnSubscribe<Integer>() {
                    @Override
                    public void call(Subscriber<? super Integer> subscriber) {
                        for (int i = 0; i < 4; i++) {
                            try {
                                Thread.sleep(i * 100);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                            subscriber.onNext(i);
                        }
                        subscriber.onCompleted();
                    }
                })
                .timeout(200, TimeUnit.MILLISECONDS, Observable.just(10, 11))
                .subscribe(new Action1<Integer>() {
                    @Override
                    public void call(Integer integer) {
                        Log.i(TAG, "call: " + integer);
                    }
                });

        // call: 0
        // call: 1
        // call: 2
        // call: 10
        // call: 11
    }
}
